package panda.transaction.recovery;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;

import panda.buffer.Buffer;
import panda.file.Block;
import panda.server.Panda;

public class RecoveryManager {
	private int transactionNumber;

	public RecoveryManager(int transactionNumber) {
		this.transactionNumber = transactionNumber;
		new StartRecord(transactionNumber).writeToLog();
	}

	public void commit() {
		Panda.getBufferManager().allPageOut(transactionNumber);
		int lsn = new CommitRecord(transactionNumber).writeToLog();
		Panda.getLogManager().flush(lsn);
	}

	public void rollback() {
		Panda.getBufferManager().allPageOut(transactionNumber);
		Iterator<LogRecord> iter = new LogRecordIterator();
		while (iter.hasNext()) {
			LogRecord record = iter.next();
			if (record.getTransactionNumber() == transactionNumber) {
				if (record.getType() == LogRecord.START)
					return;
				record.undo(transactionNumber);
			}
		}
		int lsn = new RollbackRecord(transactionNumber).writeToLog();
		Panda.getLogManager().flush(lsn);
	}

	public void recover() {
		Collection<Integer> committedTransactions = new ArrayList<Integer>();
		Iterator<LogRecord> iter = new LogRecordIterator();
		while (iter.hasNext()) {
			LogRecord record = iter.next();
			if (record.getType() == LogRecord.CHECKPOINT)
				return;
			if (record.getType() == LogRecord.COMMIT)
				committedTransactions.add(record.getTransactionNumber());
			else if (!committedTransactions.contains(record
					.getTransactionNumber()))
				record.undo(transactionNumber);
		}
		int lsn = new CheckPointRecord().writeToLog();
		Panda.getLogManager().flush(lsn);
	}

	public int setInt(Buffer buffer, int offset) {
		int value = buffer.getInt(offset);
		Block block = buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetIntRecord(transactionNumber, block, offset, value)
					.writeToLog();
	}

	public int setString(Buffer buffer, int offset) {
		String value = buffer.getString(offset);
		Block block = buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetStringRecord(transactionNumber, block, offset, value)
					.writeToLog();
	}

	public int setFloat(Buffer buffer, int offset) {
		float value = buffer.getFloat(offset);
		Block block = buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetFloatRecord(transactionNumber, block, offset, value)
					.writeToLog();
	}
	
	public int setLong(Buffer buffer, int offset){
		long value=buffer.getLong(offset);
		Block block=buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetLongRecord(transactionNumber,block,offset,value).writeToLog();
	}
	
	public int setArray(Buffer buffer, int offset){
		byte[] value=buffer.getArray(offset);
		Block block=buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetArrayRecord(transactionNumber,block,offset,value).writeToLog();
	}
	
	public int setArray(Buffer buffer, int offset,int length){
		byte[] value=buffer.getArray(offset, length);
		Block block=buffer.getBlock();
		if (block.getFileName().endsWith(".tmp"))
			return -1;
		else
			return new SetArrayRecord(transactionNumber,block,offset,value).writeToLog();
	}
}
